package com.bstek.demo.urule.cluster.mq.redis.listener;

import com.bstek.demo.urule.JsonUtils;
import com.bstek.demo.urule.cluster.mq.CustomKnowledgeCacheService;
import com.bstek.demo.urule.cluster.mq.MQConstant;
import com.bstek.demo.urule.cluster.mq.MsgObject;
import com.bstek.urule.Utils;
import com.bstek.urule.runtime.KnowledgePackage;
import com.bstek.urule.runtime.cache.CacheUtils;
import com.bstek.urule.runtime.cache.ClientCacheManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;

import java.io.IOException;
import java.nio.file.*;
import java.util.HashMap;
import java.util.Map;
import java.util.function.Consumer;

/**
 * Redis消息监听处理类，订阅到消息后的动作
 */
public class RedisConsumerListener implements MessageListener {
    Logger log = LoggerFactory.getLogger(RedisConsumerListener.class);

    private static Map<String, Consumer<String>> RULE = new HashMap<>();
    {
        RULE.put(MQConstant.CLIENT_TOPIC, this::consumerClinetMessage);
    }

    @Override
    public void onMessage(Message message, byte[] pattern) {
        byte[] b_channel = message.getChannel();
        byte[] b_body = message.getBody();
        String channel = new String(b_channel);
        String body = new String(b_body);
        log.info("channel is:{} , body is:{}", channel, body);
        RULE.get(channel).accept(body);
    }


    /**
     * 客户端（执行引擎）在订阅到消息后，处理本地缓存动作...
     * @param message
     */
    public void consumerClinetMessage(String message) {
        try {
            MsgObject msgObject = JsonUtils.fromJSON(message, MsgObject.class);
            String messageType = msgObject.getMessageType();
            String systemId = msgObject.getSystemId();
            String groupId = msgObject.getGroupId();
            String packetId = msgObject.getPacketId();
            String projectId = msgObject.getProjectId();

            log.info("messageType:{},systemId:{},groupId:{},packetId:{},projectId:{}", messageType, systemId, groupId, packetId, projectId);

            ClientCacheManager clientCacheManager = (ClientCacheManager) Utils.getApplicationContext().getBean("urule.clientCacheManager");
            CustomKnowledgeCacheService  customKnowledgeCacheService = (CustomKnowledgeCacheService) Utils.getApplicationContext().getBean(CustomKnowledgeCacheService.BEAN_ID);

            if (MQConstant.QUEUE_CLIENT_PACKET_DISABLE.equals(messageType)) {
                clientCacheManager.disableKnowledge(packetId);
            } else if (MQConstant.QUEUE_CLIENT_PACKET_ENABLE.equals(messageType)) {
                clientCacheManager.enableKnowledge(packetId);
            } else if (MQConstant.QUEUE_CLIENT_PACKET_REFRESH.equals(messageType)) {
                //clientCacheManager.reloadKnowledge(packetId);
                // 从redis中读取，再缓存到本地
                String packetStr=customKnowledgeCacheService.getCache(Long.parseLong(packetId));
                log.info("从redis中读取到的知识包信息：{}",packetStr);
                if(packetStr!=null){
                    KnowledgePackage knowledgePackage=Utils.stringToKnowledgePackage(packetStr);
                    // 写入到本地磁盘路径下
                    String dir=customKnowledgeCacheService.getKnowledgePackageFileStorePath();
                    try {
                        String filePathName=dir+"/"+packetId+".data";
                        Path filePath = Paths.get(filePathName);
                        Files.createDirectories(filePath.getParent());
                        // 创建文件
                        Files.createFile(filePath);
                        // 写入内容到文件
                        byte[] arrayOfByte = Utils.compress(packetStr);//GZIP压缩
                        Files.write(filePath, arrayOfByte, StandardOpenOption.WRITE);
                        System.out.println("文件创建成功: " + filePath);
                    } catch (IOException e) {
                        System.out.println("文件创建失败: " + e.getMessage());
                    }
                    CacheUtils.getKnowledgeCache().putKnowledge(packetId,knowledgePackage);
                    log.info("Successfully reload file packet package:{}",packetId);
                }
                // packetcache:packetcache_549
            } else if (MQConstant.QUEUE_CLIENT_JAR_SYNC.equals(messageType)) {
                // todo
                clientCacheManager.reloadDynamicJars();
                log.info("=====jar重新加载完成=====");
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error("当前客户端实例已订阅到消息，在处理消息时发生了错误：",e.getMessage());
        }
    }
}
